﻿using RabbitMQ.Client;
using System.Collections.Concurrent;
using System.Text;
using XYS.Core.Models;
using XYS.Core.Util;
using XYS.Service.Core.Consumers.Base;
using XYS.Service.Core.Models;

namespace XYS.Service.Core.Util.Helper;

/// <summary>
/// MQ 帮助类
/// </summary>
public static class MQHelper
{
    /// <summary>
    /// 配置内容
    /// </summary>
    public static RabbitMQSetting rabbitMQSetting;


    /// <summary>
    /// 发布消息用的通道
    /// </summary>
    private static readonly ConcurrentDictionary<string, IModel> channelPublishMap = new();
    /// <summary>
    /// 通道
    /// </summary>
    private static readonly ConcurrentDictionary<string, List<IModel>> channelMap = new();
    /// <summary>
    /// 连接
    /// </summary>
    private static IConnection connection = null;

    /// <summary>
    /// 初始化配置
    /// </summary>
    /// <param name="_rabbitMQSetting"></param>
    public static void Init(RabbitMQSetting _rabbitMQSetting)
    {
        rabbitMQSetting = _rabbitMQSetting;
        connection = CreateConnection();
    }

    /// <summary>
    /// 创建一个通道
    /// </summary>
    /// <param name="queueName"></param>
    /// <param name="prefetchCount"></param>
    /// <returns></returns>
    public static IModel CreateChannel(string queueName, ushort prefetchCount)
    {
        if (!channelMap.ContainsKey(queueName))
            channelMap[queueName] = new();

        var model = connection.CreateModel();
        model.BasicQos(0, prefetchCount, false);
        //声明通道 参数说明：通道名，持久性，排他性，自动删除，附加参数
        model.QueueDeclare(queueName, true, false, false, null);

        channelMap[queueName].Add(model);
        return model;
    }

    /// <summary>
    /// 获取一个发布消息用的通道
    /// </summary>
    /// <param name="queueName"></param>
    /// <returns></returns>
    public static IModel GetPublishChannel(string queueName)
    {
        if (channelPublishMap.TryGetValue(queueName, out var model) && model.IsOpen)
            return model;

        model = connection.CreateModel();
        channelPublishMap[queueName] = model;

        return model;
    }

    /// <summary>
    /// 获取通道集合
    /// </summary>
    /// <param name="queueName"></param>
    /// <returns></returns>
    public static List<IModel> GetChannels(string queueName)
    {
        if (!channelMap.TryGetValue(queueName, out List<IModel> value)) return null;
        return value;
    }
    /// <summary>
    /// 关闭并移除通道
    /// </summary>
    /// <param name="queueName"></param>
    /// <param name="channel"></param>
    public static void CloseAndRemoveChannel(string queueName, IModel channel)
    {
        if (channel == null) return;
        var channels = GetChannels(queueName);
        if (channels.Any(x => x == channel))
        {
            if (channel.IsOpen) try { channel.Close(); } catch { }
            try { channel.Dispose(); } catch { }
            channels.Remove(channel);
        }
    }


    /// <summary>
    /// 创建连接
    /// </summary>
    private static IConnection CreateConnection()
    {
        ConnectionFactory factory = new()
        {
            UserName = rabbitMQSetting.UserName,
            Password = rabbitMQSetting.Password,
            ClientProvidedName = string.IsNullOrWhiteSpace(rabbitMQSetting.ClientProvidedName) ? $"{GlobalCoreSetting.ServiceName}-MQHelper" : rabbitMQSetting.ClientProvidedName,

            DispatchConsumersAsync = true
        };
        var endpoints = rabbitMQSetting.Hosts.Select(x => new AmqpTcpEndpoint(x.Host, x.Port)).ToList();
        var connection = factory.CreateConnection(endpoints);
        return connection;
    }

    /// <summary>
    /// 获取通道
    /// </summary>
    /// <returns></returns>
    public static IModel GetChannel(string queueName, int index)
    {
        List<IModel> channels = new();
        if (channels.Count <= index) throw new Exception("【MQ】【获取一个连接通道】索引超出范围");
        return channels[index];
    }


    /// <summary>
    /// 重新创建通道
    /// </summary>
    /// <param name="consumer"></param>
    /// <param name="queueName"></param>
    public static void ReCreateChanel(this IConsumerBase consumer, string queueName)
    {
        if (connection.IsOpen)
        {
            //如果连接未被创建，则删除所有通道（重新建立连接后重新创建通道）
            if (channelMap.TryGetValue(queueName, out var channels))
            {
                foreach (var channel in channels)
                {
                    try { channel.Close(); } catch { }
                    try { channel.Dispose(); } catch { }
                }
                channels.Clear();
            }
            connection.Close();
            connection = CreateConnection();
        }
    }














    /// <summary>
    /// 锁对象
    /// </summary>
    static ConcurrentDictionary<string, object> LockMap = new();
    /// <summary>
    /// 获取锁对象
    /// </summary>
    /// <param name="key"></param>
    /// <returns></returns>
    static object GetLockObject(string key)
    {
        if (!LockMap.TryGetValue(key, out var lockObject))
        {
            lockObject = new();
            LockMap[key] = lockObject;
        }
        return lockObject;
    }
    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="exchangeName">交换器名字</param>
    /// <param name="queueName">队列名</param>
    /// <param name="routingKey">路由键</param>
    /// <param name="data"></param>
    public static void Publish<T>(string exchangeName, string queueName, string routingKey, T data)
    {
        lock (GetLockObject(queueName))
        {
            var channel = GetPublishChannel(queueName);

            //声明交换器
            if (!exchangeName.IsNullOrWhiteSpace())
                channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
            //声明通道与routingKey参数说明：通道名，持久性，排他性，自动删除，附加参数
            channel.QueueDeclare(queueName, true, false, false, null);
            //通道绑定交换器
            if (!exchangeName.IsNullOrWhiteSpace())
                channel.QueueBind(queueName, exchangeName, routingKey, null);

            byte[] messageBodyBytes = Encoding.UTF8.GetBytes(data.ToJson());
            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";

            //持久化消息
            props.DeliveryMode = 2;
            //props.Expiration = "36000000";

            channel.BasicPublish(exchangeName, routingKey, props, messageBodyBytes);
        }
    }

    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="queueName">队列名</param>
    /// <param name="data"></param>
    public static void Publish<T>(string queueName, T data)
    {
        lock (GetLockObject(queueName))
        {
            var channel = GetPublishChannel(queueName);

            string exchangeName = "";
            string routingKey = queueName;
            //声明通道与routingKey参数说明：通道名，持久性，排他性，自动删除，附加参数
            channel.QueueDeclare(queueName, true, false, false, null);

            byte[] messageBodyBytes;
            if (typeof(T) == typeof(string))
                messageBodyBytes = Encoding.UTF8.GetBytes(data.ToString());
            else
                messageBodyBytes = Encoding.UTF8.GetBytes(data.ToJson());

            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";

            //持久化消息
            props.DeliveryMode = 2;
            channel.BasicPublish(exchangeName, routingKey, props, messageBodyBytes);
        }
    }


    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="exchangeName">交换器名字</param>
    /// <param name="queueName">队列名</param>
    /// <param name="routingKey">路由键</param>
    /// <param name="data"></param>
    public static void PublishBatch<T>(string exchangeName, string queueName, string routingKey, List<T> data)
    {
        lock (GetLockObject(queueName))
        {
            var channel = GetPublishChannel(queueName);

            //声明交换器
            if (!exchangeName.IsNullOrWhiteSpace())
                channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, true);
            //声明通道与routingKey参数说明：通道名，持久性，排他性，自动删除，附加参数
            channel.QueueDeclare(queueName, true, false, false, null);
            //通道绑定交换器
            if (!exchangeName.IsNullOrWhiteSpace())
                channel.QueueBind(queueName, exchangeName, routingKey, null);

            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";

            //持久化消息
            props.DeliveryMode = 2;

            var batch = channel.CreateBasicPublishBatch();
            foreach (var item in data)
            {
                ReadOnlyMemory<byte> body = Encoding.UTF8.GetBytes(item.ToJson());
                batch.Add(exchangeName, routingKey, false, props, body);
            }
            batch.Publish();
        }
    }

    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="queueName">队列名</param>
    /// <param name="data"></param>
    public static void PublishBatch<T>(string queueName, List<T> data)
    {
        lock (GetLockObject(queueName))
        {
            var channel = GetPublishChannel(queueName);
            string exchangeName = "";
            string routingKey = queueName;
            //声明通道与routingKey参数说明：通道名，持久性，排他性，自动删除，附加参数
            channel.QueueDeclare(queueName, true, false, false, null);

            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";

            //持久化消息
            props.DeliveryMode = 2;

            var batch = channel.CreateBasicPublishBatch();
            foreach (var item in data)
            {
                ReadOnlyMemory<byte> body = Encoding.UTF8.GetBytes(item.ToJson());
                batch.Add(exchangeName, routingKey, false, props, body);
            }
            batch.Publish();
        }
    }



    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="exchangeName">交换器名字</param>
    /// <param name="queueName">队列名</param>
    /// <param name="routingKey">路由键</param>
    /// <param name="data"></param>
    /// <param name="delayTime"></param>
    public static void Publish<T>(string exchangeName, string queueName, string routingKey, T data, TimeSpan delayTime)
    {
        lock (GetLockObject(queueName))
        {
            IModel publishChannel = GetPublishChannel(queueName);

            // 声明原始交换机和队列（用于最终消费）
            if (!exchangeName.IsNullOrWhiteSpace())
                publishChannel.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: true);
            publishChannel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, null);
            if (!exchangeName.IsNullOrWhiteSpace())
                publishChannel.QueueBind(queueName, exchangeName, routingKey, null);

            // 声明延迟队列（添加死信交换机配置）
            string delayQueueName = $"{queueName}_delay";
            var delayQueueArgs = new Dictionary<string, object>
{
    { "x-dead-letter-exchange", exchangeName },       // 死信转发到原始交换机
    { "x-dead-letter-routing-key", routingKey }       // 使用原始路由键
};
            publishChannel.QueueDeclare(
                queue: delayQueueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: delayQueueArgs
            );

            // 准备消息属性
            IBasicProperties basicProperties = publishChannel.CreateBasicProperties();
            basicProperties.ContentType = "application/json";
            basicProperties.DeliveryMode = 2;
            basicProperties.Expiration = delayTime.TotalMilliseconds.ToString(); // 设置消息TTL（毫秒）

            byte[] messageBodyBytes = Encoding.UTF8.GetBytes(data.ToJson());
            publishChannel.BasicPublish(delayQueueName, routingKey, basicProperties, messageBodyBytes);
        }
    }
    /// <summary>
    /// 发布消息
    /// </summary>
    /// <param name="queueName">队列名</param>
    /// <param name="data"></param>
    /// <param name="delayTime"></param>
    public static void Publish<T>(string queueName, T data, TimeSpan delayTime)
    {
        lock (GetLockObject(queueName))
        {
            IModel publishChannel = GetPublishChannel(queueName);

            publishChannel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, null);

            // 声明延迟队列（添加死信交换机配置）
            string delayQueueName = $"{queueName}_delay";
            var delayQueueArgs = new Dictionary<string, object>
{
    { "x-dead-letter-exchange", "" },       // 死信转发到原始交换机
    { "x-dead-letter-routing-key", queueName }       // 使用原始路由键
};
            publishChannel.QueueDeclare(
                queue: delayQueueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: delayQueueArgs
            );

            // 准备消息属性
            IBasicProperties basicProperties = publishChannel.CreateBasicProperties();
            basicProperties.ContentType = "application/json";
            basicProperties.DeliveryMode = 2;
            basicProperties.Expiration = delayTime.TotalMilliseconds.ToString(); // 设置消息TTL（毫秒）

            byte[] messageBodyBytes = Encoding.UTF8.GetBytes(data.ToJson());
            publishChannel.BasicPublish("", delayQueueName, basicProperties, messageBodyBytes);
        }
    }

    /// <summary>
    /// 发布延迟队列
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="exchangeName"></param>
    /// <param name="queueName"></param>
    /// <param name="routingKey"></param>
    /// <param name="data"></param>
    /// <param name="delayTime"></param>
    public static void PublishBatch<T>(string exchangeName, string queueName, string routingKey, List<T> data, TimeSpan delayTime)
    {
        lock (GetLockObject(queueName))
        {
            IModel publishChannel = GetPublishChannel(queueName);

            // 声明原始交换机和队列（用于最终消费）
            if (!exchangeName.IsNullOrWhiteSpace())
                publishChannel.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: true);
            publishChannel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, null);
            if (!exchangeName.IsNullOrWhiteSpace())
                publishChannel.QueueBind(queueName, exchangeName, routingKey, null);

            // 声明延迟队列（添加死信交换机配置）
            string delayQueueName = $"{queueName}_delay";
            var delayQueueArgs = new Dictionary<string, object>
{
    { "x-dead-letter-exchange", exchangeName },       // 死信转发到原始交换机
    { "x-dead-letter-routing-key", routingKey }       // 使用原始路由键
};
            publishChannel.QueueDeclare(
                queue: delayQueueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: delayQueueArgs
            );

            // 准备消息属性
            IBasicProperties basicProperties = publishChannel.CreateBasicProperties();
            basicProperties.ContentType = "application/json";
            basicProperties.DeliveryMode = 2;
            basicProperties.Expiration = delayTime.TotalMilliseconds.ToString(); // 设置消息TTL（毫秒）

            // 批量发布到延迟队列
            IBasicPublishBatch batch = publishChannel.CreateBasicPublishBatch();
            foreach (T datum in data)
            {
                ReadOnlyMemory<byte> body = Encoding.UTF8.GetBytes(datum.ToJson());
                batch.Add(
                    exchange: "",                   // 使用默认直连交换机
                    routingKey: delayQueueName,     // 直接路由到延迟队列
                    mandatory: false,
                    properties: basicProperties,
                    body: body
                );
            }
            batch.Publish();
        }
    }
    /// <summary>
    /// 发布延迟队列
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName"></param>
    /// <param name="data"></param>
    /// <param name="delayTime"></param>
    public static void PublishBatch<T>(string queueName, List<T> data, TimeSpan delayTime)
    {
        // 使用默认交换机（空字符串）和队列名称作为路由键
        PublishBatch("", queueName, queueName, data, delayTime);
    }



    /// <summary>
    /// 获取消息
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName">队列名</param>
    /// <param name="messageCount">获取多少数据</param>
    /// <param name="autoAck">设置为true表示自动确认，false表示需要手动确认</param>
    /// <returns></returns>
    public static List<T> GetMessage<T>(string queueName, int messageCount, bool autoAck)
    {
        //获取一个 channel 
        using var channel = connection.CreateModel();
        //声明通道 参数说明：通道名，持久性，排他性，自动删除，附加参数
        channel.QueueDeclare(queueName, true, false, false, null);

        List<T> result = new();
        for (int i = 0; i < messageCount; i++)
        {
            // 获取消息
            BasicGetResult basicGetResult = channel.BasicGet(queueName, autoAck);
            if (basicGetResult == null) return default;

            var body = basicGetResult.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            result.Add(message.ToObject<T>());
        }
        return result;
    }
}
